package com.emc.greenplum.pigsterix.compiler.rules;


import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.mutable.Mutable;
import org.apache.commons.lang3.mutable.MutableObject;

import com.emc.greenplum.pigsterix.compiler.TaggingState;
import com.emc.greenplum.pigsterix.compiler.PigConstantValue;
import com.emc.greenplum.pigsterix.compiler.PigFunctionInfo;
import com.sun.tools.javac.util.Pair;

import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalExpression;
import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalOperator;
import edu.uci.ics.hyracks.algebricks.core.algebra.base.IOptimizationContext;
import edu.uci.ics.hyracks.algebricks.core.algebra.base.IPhysicalOperator;
import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalOperatorTag;
import edu.uci.ics.hyracks.algebricks.core.algebra.base.LogicalVariable;
import edu.uci.ics.hyracks.algebricks.core.algebra.base.PhysicalOperatorTag;
import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.ScalarFunctionCallExpression;
import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.VariableReferenceExpression;
import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.AbstractLogicalOperator;
import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.AssignOperator;
import edu.uci.ics.hyracks.algebricks.core.algebra.operators.logical.ExchangeOperator;
import edu.uci.ics.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
import edu.uci.ics.hyracks.api.constraints.expressions.ConstantExpression;

public class ConvertAlgebricks2MapReduceRule implements IAlgebraicRewriteRule {

	private String tagStateAnnotationKey = ConvertAlgebricks2MapReduceRule.class.getName()+".LabelProvider";
    @Override
    public boolean rewritePre(Mutable<ILogicalOperator> opRef, IOptimizationContext context) throws AlgebricksException {

		AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();
		IPhysicalOperator opPhy = op.getPhysicalOperator();
		PhysicalOperatorTag tag = opPhy.getOperatorTag();

		Map<String, Object> annotations = op.getAnnotations();
		Pair tag_parent = (Pair)annotations.get(tagStateAnnotationKey);
		//at the root level, if the annotation doesnt exist - create it
		if (tag_parent == null) {
			tag_parent = new Pair (new TaggingState(), 0);
			annotations.put(tagStateAnnotationKey, tag_parent);
		} 
		System.out.println(op.toString() +" __#__"+ ((TaggingState)tag_parent.fst).getLabel() + tag_parent.snd);
		//iterate on the children and point to the parent annotation
		List<Mutable<ILogicalOperator>> children_list= op.getInputs();
		
		for(Mutable<ILogicalOperator> child:children_list){
			AbstractLogicalOperator child_sp = (AbstractLogicalOperator) child.getValue();
			Map<String, Object> annotations_child  = child_sp.getAnnotations();
			
			IPhysicalOperator opPhy1 = child_sp.getPhysicalOperator();
			PhysicalOperatorTag tag1 = opPhy1.getOperatorTag();
			
			boolean exchange = (tag1 == PhysicalOperatorTag.HASH_PARTITION_EXCHANGE) && 
			   (tag1 == PhysicalOperatorTag.HASH_PARTITION_EXCHANGE);
			
			if (exchange){
				((TaggingState)(tag_parent.fst)).incrementAndSetLabel();
				Pair p = new Pair(tag_parent.fst, ((TaggingState)tag_parent.fst).getLabel());
				annotations_child.put(tagStateAnnotationKey, p);
				System.out.println("exchange __#__" + ((TaggingState)p.fst).getLabel()+ p.snd);
			
			}
			else{
				annotations_child.put(tagStateAnnotationKey, tag_parent);
			}
			
		}
  	
    	return false;
    }

    @Override
    public boolean rewritePost(Mutable<ILogicalOperator> opRef, IOptimizationContext context)
            throws AlgebricksException {
    	
    	//Building the physical operator back from the generic parameter value
    	AbstractLogicalOperator op = (AbstractLogicalOperator) opRef.getValue();
    	
		if (op.getOperatorTag() == LogicalOperatorTag.EXCHANGE) {
		
			IPhysicalOperator opPhy = op.getPhysicalOperator();
			PhysicalOperatorTag tag = opPhy.getOperatorTag();

			//wrap the subtree into a Map
			if (tag == PhysicalOperatorTag.HASH_PARTITION_EXCHANGE) {
				
				//Extract the keys
				ExchangeOperator exOp = (ExchangeOperator) op;
								
				LogicalVariable var1 = context.newVar(); //key
				LogicalVariable var2 = context.newVar(); //value
				
				List<LogicalVariable> columns = new ArrayList<LogicalVariable>();
				exOp.getPhysicalOperator().getDeliveredProperties().getPartitioningProperty().getColumns(columns);
				List<Mutable<ILogicalExpression>> varRefs1 = new ArrayList<Mutable<ILogicalExpression>>();
				for (LogicalVariable var: columns){
					varRefs1.add(new MutableObject(new VariableReferenceExpression(var)));
				}
				ScalarFunctionCallExpression expr1 = new ScalarFunctionCallExpression(PigFunctionInfo.COMP_KEY, varRefs1);
				
								
				List<LogicalVariable> schema = exOp.getInputs().get(0).getValue().getSchema();
				List<Mutable<ILogicalExpression>> varRefs2 = new ArrayList<Mutable<ILogicalExpression>>();
				varRefs2.add(new MutableObject(new ConstantExpression(new PigConstantValue(new String(), Integer.TYPE))));
				for (LogicalVariable var: schema){
					varRefs2.add(new MutableObject(new VariableReferenceExpression(var)));
				}
				ScalarFunctionCallExpression expr2 = new ScalarFunctionCallExpression(PigFunctionInfo.PACK_VALUE, varRefs2);
				
				
				List<LogicalVariable> vars  = Arrays.asList(var1, var2);
				List<Mutable<ILogicalExpression>> exprs = Arrays.<Mutable<ILogicalExpression>>asList(new MutableObject<ILogicalExpression>(expr1), new MutableObject<ILogicalExpression>(expr2));
				
				AssignOperator assign_key = new AssignOperator(vars, exprs);
				
				
			}
			return false;
		}
    	return false;
    }
}

